package com.yxc.emqxdemo.configuration;

import com.yxc.emqxdemo.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.Set;

/**
 * emqx的客户端
 *
 * @author: YXC
 * @time: 2022/9/15 11:17
 **/
@Slf4j
@Configuration
public class EmqxClient {

    @Autowired
    private EmqxProperties properties;

    @Autowired
    private EmqxCallback emqxCallback;

    @Autowired
    private MqttClient emqxConnect;

    @Bean(name = "emqxConnect")
    public MqttClient connect() {
        MqttClient client = null;
        try {
            log.info("正在创建emqx连接");
            // 创建客户端
            client = new MqttClient(properties.getHost(), properties.getClientId(), new MemoryPersistence());
            // 连接参数
            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(properties.getUsername());
            options.setPassword(properties.getPassword().toCharArray());
            options.setConnectionTimeout(properties.getTimeout());
            options.setKeepAliveInterval(properties.getKeepalive());
            // 保留会话
            options.setCleanSession(true);
            // 自动重连
            options.setAutomaticReconnect(true);
            // 设置回调
            client.setCallback(emqxCallback);
            // 建立连接
            client.connect(options);
            log.info("emqx连接建立成功");
            setTopic(client);
        } catch (MqttException e) {
            e.printStackTrace();
        }
        return client;
    }

    /**
     * 10秒检查一次  检查emqx是否连接
     */
    @Scheduled(fixedDelay = 10000)
    public void checkAlive() throws MqttException {

        if (!this.emqxConnect.isConnected()) {
            log.error("emqx连接中断，重新连接中。。。");
            this.reconnect();
        }
    }

    private void setTopic(MqttClient client) {
        // 订阅topic
        try {
            Set<String> topics = emqxCallback.getConsumerBean().keySet();
            if (!topics.isEmpty()) {

                while (!client.isConnected()) {
                    Thread.sleep(1000);
                }
                client.subscribe(topics.toArray(new String[topics.size()]));
                log.info("订阅topic: [{}] 成功", StringUtils.collectionToString(topics, ","));
            }
        } catch (MqttException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void reconnect() throws MqttException {
        this.emqxConnect.reconnect();
        setTopic(this.emqxConnect);
        log.info("已经重新连接");
    }

    /**
     * 发送消息
     *
     * @author YXC
     * @date 2022/9/15 15:24
     * @param topic: 发送的topic
     * @param pushMessage: 发送的消息
     * @return void
    **/
    public void publish(String topic, String pushMessage) {
        publish(true, topic, pushMessage, emqxConnect);
    }

    /**
     * 发送消息
     *
     * @author YXC
     * @date 2022/9/15 15:12
     * @param retained: 服务器是否需要保留
     * @param topic: 发送的topic
     * @param pushMessage: 发送的消息
     * @param emqxConnect: emqx连接
     * @return void
    **/
    private void publish(boolean retained, String topic, String pushMessage, MqttClient emqxConnect) {
        MqttMessage message = new MqttMessage();
        message.setQos(properties.getQos());
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        try {
            emqxConnect.publish(topic, message);
            log.debug("消息发送成功{}: {}", topic, pushMessage);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

}
